package task6

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.Expressions.$
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, dataSetConversions}
import org.apache.flink.types.Row

import scala.tools.util.PathResolver.Environment

/**
 * @author jhhe66
 * @date 2021/6/18 11:57
 */
object TableDemo {
  def main(args: Array[String]): Unit = {
    def main(args: Array[String]): Unit = {
      // 配置环境参数
      val settings = EnvironmentSettings.newInstance()
        .useBlinkPlanner()
        .inStreamingMode()
        .build()

      // 获取流环境
      val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
      // 创建表环境
      val tEnv = StreamTableEnvironment.create(env, settings)
      // 根据socket生成kv的元组
      val dataDS: DataStream[(String, Int)] = env.socketTextStream("centos001", 9999)
        .map { line =>
          var arr = line.split(",")
          (arr(0).trim, arr(1).trim.toInt)
        }
      // 根据流对象生成表，指定字段名
      tEnv.createTemporaryView("t", dataDS, $("name"), $("num"))
      // 定义sql, 求每个key的num和
      val sql = "select name, sum(num) as sum_num from t group by name"
      // 执行sql，返回一个表对象
      val tableRes: Table = tEnv.sqlQuery(sql)
      val res: DataStream[(Boolean, Row)] = tEnv.toRetractStream[Row](tableRes)
      res.print()
      env.execute()
    }
  }
}
